//#include "gateway_error.h"
//#include "gateway_packet.h"
#include "gateway_client.h"
#include <boost/regex.hpp>
#include <boost/format.hpp>
#include <boost/lexical_cast.hpp>  
#include <boost/make_shared.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/ini_parser.hpp>
#include <boost/property_tree/json_parser.hpp>
#include <string.h>

#include "../tool/crc16.h"
#include "../tool/encrpty.h"
#include "../tool/cbase64.h"
#include "../pagefile/pagefile.h"
#include "../mqttservice/mqttservice.h"
#include "../tool/md5.h"
#include "../../liblog/Include/liblog.h"
#include "gateway_taskpool.h"
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/date_time.hpp>  
#include "../mqttservice/mqttservice.h"
#include "gateway_cache.h"
#include <iostream>  
#include <sstream>
/*
struct _typeText
{
    const unsigned char type;
    const char* meaning;
};

struct _nameText
{
    const char* name;		//属性名
    const char* type;		//属性类型
    const char* desc;		//属性描述
    _typeText typeText[3];
};
static const _nameText g_NameText[] =
{
    {"vstate", "enum", "阀门状态", {{0, "open"}, {1, "close"}, {2, "abnormal"}}},
    {"ovva",   "bool", "关阀电压告警",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"pmark",  "bool", "提示标记",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"alarm",  "bool", "报警状态",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"mstate", "bool", "余量状态",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"cct",    "bool", "计数通道二",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"cco",    "bool", "计数通道一",{{0, "false"}, {1, "true"}, {0xff, ""}}},

    {"wmstatus",   "bool", "无线模块状态",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"smisign",    "bool", "强磁干扰标志",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"foffvmark",  "bool", "强制关阀标记",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"cfsign",     "bool", "电路故障标志",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"fopenvmark", "bool", "强制开阀标记",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"ovamark",    "bool", "关阀报警标记",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"mmark",      "bool", "余量标记",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"omark",      "bool", "透支标记",{{0, "false"}, {1, "true"}, {0xff, ""}}},

    {"sfasign",   "bool", "小流量异常标志",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"lfasign",   "bool", "大流量异常标志",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"glmark",   "bool", "燃气泄漏标志",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"dytbsign",   "bool", "干黄管坏标志",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"pvalarm",   "bool", "提示电压报警",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"csstatus",   "bool", "插卡开关状态",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    {"getstatus",   "bool", "查询状态",{{0, "false"}, {1, "true"}, {0xff, ""}}},
    //{"ovva",   "bool", "关阀电压告警",{{0, "false"}, {1, "true"}, {0xff, ""}}},

};
*/
using namespace boost::property_tree;
using namespace boost::gregorian;
using namespace boost::posix_time;
using namespace boost;
using namespace std;

const int time_out = 300;// s

gateway_client::gateway_client(gateway_stream_ptr stream_ptr)
 :_is_connected(false)
 ,_stream_ptr(stream_ptr)
 ,_mqtt(NULL) 
 ,_protocol("", 0){
}

gateway_client::~gateway_client() {
    if (_mqtt)
        delete _mqtt;
    _mqtt = NULL;
    //printf("%ld New client[%s:%u] ~gateway_client().\n", time(NULL), _host.c_str(), _port);
}

/**** gateway_stream_event ****/
int gateway_client::stream_init() {
	_host = _stream_ptr->get_remote_host();
    _port = _stream_ptr->get_remote_port();
    _addr = _host + std::string(":") +boost::lexical_cast<std::string>(_port);
    _protocol.init(_host, _port);
    get_mqtt_config(_mqttconfig);
    _mqtt = new CMQTTService(this);
    
	//printf("%ld New client[%s] connected.\n", time(NULL), _addr.c_str());
    _client_ptr.reset(this);
	return 0;
}

int gateway_client::stream_clean() {
	//printf("%ld Client[%s:%u] disconnected.\n", time(NULL), _host.c_str(), _port);
    if (_mqtt)
    {
            _mqtt->Stop();
            delete _mqtt;
            _mqtt = NULL;
    }
    gateway_cache::get_instance()->del_client(_strdevid, _addr);
	_client_ptr.reset();
	return 0;
}

int gateway_client::stream_read(unsigned char* data, unsigned int dataLen) {
    unsigned int used = 0;
    _stream_ptr->set_timeout(30);//超时30s
	int ret = handle_parse_packet(data, dataLen, used);
	if (ret < 0)
	{
		//log
		_stream_ptr->close();
		return -1;
	}
	return std::min(dataLen, used);
}

/**** gateway_task ****/
int gateway_client::on_task(void* param) {
	if (param == NULL)
		return -1;
    if (_mqtt == NULL)
    return -1;
    event_data *ped = (event_data*)param;
    if (ped->tasktype == PULISH)
    {
        int ret = 0;
        pulish_data *ppdata = (pulish_data*)(ped->data);
        std::string strjson = std::string(ppdata->data, ppdata->datalen);
        delete[] ppdata->data;
        delete ppdata;
        ret = _mqtt->pushUpdateData(strjson.length(), (const void*)strjson.c_str());
        if (ret)
        {
            //log
            LogError(("%s-mqttpush ret:%d, val:%s\r\n", __PRETTY_FUNCTION__, ret, strjson.c_str()));            
        }
    }
    else if (ped->tasktype == SUBSCRIBE)
    {
        onmessage_data *pmsg = (onmessage_data*)(ped->data);
        task_onmessage(pmsg->strid, pmsg->actions, pmsg->payload);
        delete pmsg;
    }
    else
    {
        _mqtt->devSubscribe();
        //执行get操作
        _mqtt->pushGetData();
    }
    delete ped; 
    return 0;
}


int gateway_client::handle_parse_packet(unsigned char* buf, unsigned int bufLen, unsigned int &used) {
    used = 0;
    unsigned int total = bufLen;
    unsigned int pos = 0;   
    std::string protoText((char*)(buf + used), total);
    LogDebug(("RCV:%s\r\n", protoText.c_str()));      
    pos = protoText.find("\n", used);
    if (pos <= 0)
        return 0;

    do{

        int enterpos = protoText.find("\n", used);
        if (enterpos <= 0)
            break;//完整的命令解析完了
        std::string cmddata = protoText.substr(used, (++enterpos - used));
        used = enterpos;
        if (_protocol.prase((const unsigned char*)cmddata.c_str(), cmddata.length()) < 0)
        {
            //log
            continue;
        }
        
    }while (total - used > 0);
    _strdevid = _protocol.get_devid();
    std::string strjsonbuf = _protocol.get_json_buf();
    if (_strdevid.length() == 0 || strjsonbuf.length() == 0)
        return -1;
	
    return handle_mqtt_event(strjsonbuf);
}

int gateway_client::handle_mqtt_event(std::string strjson)
{
	//获取mqtt id和key
	if (_mqtt->IsConnect() == false &&  _is_connected == false)//没有连接
	{
		std::string strdeviceid = "";
		std::string strdevicekey = "";
		if (!CPagefile::GetInstance().GetNewDeviceInfo(_strdevid, strdeviceid, strdevicekey))
			return -1;

		MD5 md5(strdeviceid + strdevicekey);
		_mqttconfig.strusrname = strdeviceid;
		_mqttconfig.strmqttid = strdeviceid;
		_mqttconfig.strusrpwd = md5.hexdigest();
        _mqttconfig.strkey = strdevicekey;
        
        //
        LogDebug(("%s:mqttid- %s\r\n",  _strdevid.c_str(), _mqttconfig.strmqttid.c_str()));
       //
       gateway_cache::get_instance()->insert_client(_strdevid, _client_ptr);
		if (_mqtt->Start(_mqttconfig))
			return -1;		

		//if (_mqtt->devSubscribe())
		//{
		//	_mqtt->disconnect();
		//	return -1;		
        //};
        
        _is_connected = true;
	}
    event_data *ed = new event_data;
    ed->tasktype = PULISH;
    pulish_data *pdata = new pulish_data;
    pdata->data = new char[strjson.length()];
    strjson.copy(pdata->data, strjson.length(), 0);
    pdata->datalen = strjson.length();
    ed->data = (void*) pdata;
    gateway_taskpool::get_instance()->schedule(boost::static_pointer_cast<gateway_task>(_client_ptr), (void*)ed);
    return 0;
}

int gateway_client::gateway_packet_ack(){
    return 0;
}

int gateway_client::get_mqtt_config(MQTTCONFIG & mqttconfig)
{
    
	std::string zk_hosts;
	try
	{
		boost::property_tree::ptree reader;
		boost::property_tree::ini_parser::read_ini("collectionsvr.ini", reader);
		mqttconfig.strmqtthost = reader.get<std::string>("mqtt.HOSTS");
        mqttconfig.mqttport = reader.get<int>("mqtt.PORT");
	}
	catch (std::exception& ec)
	{
		mqttconfig.strmqtthost = "127.0.0.1";
		mqttconfig.mqttport = 1883;
		LogError(("%s failed.error:%s.\r\n", __PRETTY_FUNCTION__, ec.what()));
		return -1;
	}
    return 0;
}

/*
boost::shared_ptr<gateway_client> gateway_client::get_ptr()
{
    //return boost::dynamic_pointer_cast<gateway_client> (shared_from_this());
    return shared_from_this();
}
*/


int gateway_client::OnConnect(CMQTTService *mq, std::string strid, int rc)
{
    if (rc != 0)
    {
        //连接失败
		LogError(("%s:%s连接MQTT服务器失败,返回错误码:%d\r\n", __PRETTY_FUNCTION__, strid.c_str(), rc));
        return rc;
    }
    
    LogDebug(("%s 连接mqtt服务器成功\r\n", strid.c_str()));
    event_data *ped = new event_data;
    ped->tasktype =  ONCONNECT;
    ped->data = NULL;
    gateway_taskpool::get_instance()->schedule(boost::static_pointer_cast<gateway_task>(_client_ptr), (void*)ped);
    ///mq->devSubscribe();

    //执行get操作
    //mq->pushGetData();
    return 0;
}
int gateway_client::OnDisconnect(CMQTTService *mq, std::string strid, int rc)
{
	LogDebug(("%s:断开,返回码%d \r\n", strid.c_str(), rc));
    if (!mq)
        return 0;
    //mq->reconnect();
    
    return 0;
}
int gateway_client::OnMessage(CMQTTService *mq, std::string strid, ShadowActions actions, std::string payload)
{
	//if (!_stop)
    //_threadpool.schedule(boost::bind(&CTransaction::DealCommandInfo, this, strid, actions, payload));
    event_data *ed = new event_data;
    ed->tasktype = SUBSCRIBE;
    onmessage_data *pmsgdata = new onmessage_data;
    pmsgdata->strid = strid;
    pmsgdata->actions = actions;
    pmsgdata->payload = payload;
    ed->data = (void*)pmsgdata; 
    gateway_taskpool::get_instance()->schedule(boost::static_pointer_cast<gateway_task>(_client_ptr), (void*)ed);
    //LogDebug(("%s 收到消息%s\r\n", strid.c_str(), payload.c_str()));
    return 0;
}
int gateway_client::OnSubscribe(CMQTTService *mq, std::string strid, int mid, int qos_count, const int *granted_qos)
{
    LogDebug(("%s 订阅topic成功成功\r\n", strid.c_str()));
    return 0;
}

int gateway_client::OnUnsubscribe(CMQTTService *mq, std::string strid, int mid)
{
    return 0;
}

int gateway_client::OnPublish(CMQTTService *mq, std::string strid, int mid)
{
    LogDebug(("%s: 上报成功\r\n", strid.c_str()));
    return 0;
}


void gateway_client::close()
{
    _stream_ptr->close();
}

int gateway_client::task_pulish(std::string strpulishdata)
{
    int ret = 0;
    if (_mqtt == NULL)
        return -1;

    if (strpulishdata.length() <= 0)
        return -1;

    ret = _mqtt->pushUpdateData(strpulishdata.length(), (const void*)strpulishdata.c_str());
    if (ret)
    {
        //log
        LogError(("%s-mqttpush ret:%d, val:%s\r\n", __PRETTY_FUNCTION__, ret, strpulishdata.c_str()));
    }
    return ret;
}


void gateway_client::task_onmessage(std::string strid, ShadowActions actions, std::string strpayload)
{
    int ret = 0;
    switch (actions)
    {
    case SHADOW_UPDATE:
    {
        if ((ret = deal_update(strpayload)) != 0)
            LogError(("%s: 设备更新影子失败 错误%d\r\n", strid.c_str(), ret));
        break;
    }
    case SHADOW_GET:
    {
        if (deal_get(strid, strpayload) != 0)
            LogError(("%s: 设备主动获取影子数据%s处理失败\r\n", strid.c_str(), strpayload.c_str()));
        break;
    }
    case SHADOW_CTRL:
    {
        if (deal_ctrl(strid, strpayload) != 0)
            LogError(("%s: 设备控制命令下发处理失败\r\n", strid.c_str()));
        break;
    }
    default:
        return;
    }
    return;

}

std::string gateway_client::get_json_val(const std::string strJson, const std::string strkey)
{
    std::stringstream stream(strJson);
    boost::property_tree::ptree pt;
    std::string strjsonval = "";
    try{  
        read_json<boost::property_tree::ptree>(stream, pt);
        strjsonval = pt.get<std::string>(strkey); 
    }catch(boost::property_tree::ptree_error & e) {  
        LogError(("%s: %s json form error\r\n", __PRETTY_FUNCTION__, strJson.c_str()));
        strjsonval = "";   
    }
    return strjsonval;

}

int gateway_client::deal_update(const std::string strJson)
{
    std::string strstatus = get_json_val(strJson, "status");
    if (strstatus.length() == 0)
        return -1;

    int istat = boost::lexical_cast<int>(strstatus);
    if (istat != 200)
    {
        return istat;
    }
    return 0;
}

int gateway_client::deal_get(const std::string strdevid, const std::string strJson)
{
    std::string strstatus = get_json_val(strJson, "status");
    if (strstatus.length() == 0)
        return -1;
    
    int istat = boost::lexical_cast<int>(strstatus);
    if (istat != 200)
    {
        return -1;
    }

    stringstream stream(strJson);
    ptree pt, ptresponse,ptstate, ptdesired, ptreported, ptmetadata, ptdesiredtime;
    std::string strjsonval = "";
    try{  
        read_json<ptree>(stream, pt);
        ptresponse = pt.get_child("response");
        ptmetadata = ptresponse.get_child("metadata");
        ptstate = ptresponse.get_child("state"); 
        ptreported = ptstate.get_child("reported");
    }catch(ptree_error & e) {  
    
    }

    try{  
        ptdesired = ptstate.get_child("desired");
        ptdesiredtime = ptmetadata.get_child("desired");
    }catch(ptree_error & e) {  
        LogError(("%s: %s json not exit \"desired\" key\r\n", __PRETTY_FUNCTION__, strJson.c_str()));
        return 0;
    }

    ptree ptmp;
    int timestamp = 0;
    for (boost::property_tree::ptree::iterator it=ptdesired.begin(); it!=ptdesired.end(); it++)  
    {  
        //遍历读出数据  
        std::string key=it->first;//key ID  
        std::string val=it->second.data();  
        std::string dstval = "";
        try{
            dstval = ptreported.get<std::string>(key);
        }
        catch(ptree_error & e) {  
            std::stringstream bodystream;
            bodystream.clear();
            write_json(bodystream, ptreported);
            std::string strbody = bodystream.str();
            LogError(("%s: %s json get key \"%s\" value error\r\n", __PRETTY_FUNCTION__,  strbody.c_str(), key.c_str()));
            //continue;
        }
        if (val.compare(dstval) == 0)
        {
            LogDebug(("%s: 命令%s 与影子状态%s相同,不响应\r\n", strdevid.c_str(), key.c_str(),val.c_str()));
            continue;//下发命令与状态相同，不响应
        }

        //判断时间是否超时
        try{
            ptmp = ptdesiredtime.get_child(key);
            timestamp = ptmp.get<int>("timestamp");
        }catch(ptree_error & e) {  
            std::stringstream bodystream;
            bodystream.clear();
            write_json(bodystream, ptmp);
            std::string strbody = bodystream.str();
            LogError(("%s: %s json get key \"%s\" val error\r\n", __PRETTY_FUNCTION__,  strbody.c_str(), key.c_str()));
            continue;
        }
        
        int timeutc = get_current_stamp();
        if (timeutc - timestamp > time_out)//命令超过当前5分钟则掉丢
        {
            LogDebug(("%s: 命令%s 超时,不响应\r\n", strdevid.c_str(), key.c_str()));
            continue;
        }
        //下发命令
        std::string strcmd = _protocol.get_ctrcmddata(key, val);
        if (strcmd.length() == 0)
            continue;
        //下发命令
        if (send_cmd(strcmd) != (int)strcmd.length())
        {
            LogError(("%s: %s send cmd %s error\r\n", __PRETTY_FUNCTION__, strdevid.c_str(), strcmd.c_str()));
        }

        //设置超时时间
        int time  = _protocol.get_retime();
        if (time > 0)
        _stream_ptr->set_timeout(time * 2);//超时时间间隔的两倍
        
    }   
    return 0;
}


int gateway_client::deal_ctrl(const std::string strdevid, const std::string strJson)
{
  
    stringstream stream(strJson);
    ptree pt, ptstate, ptdesired, ptreported;
    std::string strjsonval = "";
    try{  
        read_json<ptree>(stream, pt);
        ptstate = pt.get_child("state"); 
        ptdesired = ptstate.get_child("desired");
        ptreported = ptstate.get_child("reported");
    }catch(ptree_error & e) {  
        LogError(("%s: %s json form error\r\n", __PRETTY_FUNCTION__, strJson.c_str()));
        strjsonval = "";   
    }
    
    for (boost::property_tree::ptree::iterator it=ptdesired.begin(); it!=ptdesired.end(); it++)  
    {  
        //遍历读出数据  
        std::string key=it->first;//key ID  
        std::string val=it->second.data();
        std::string dstval = "";
        try{
            dstval = ptreported.get<std::string>(key);
        }catch(ptree_error & e){
            std::stringstream bodystream;
            bodystream.clear();
            write_json(bodystream, ptreported);
            std::string strbody = bodystream.str();
            LogError(("%s: %s json get key \"%s\" value error\r\n", __PRETTY_FUNCTION__,  strbody.c_str(), key.c_str()));
            //continue;
        }
        
        if (val.compare(dstval) == 0)
        {
            LogDebug(("%s: 命令%s 与影子状态%s相同,不响应\r\n", strdevid.c_str(), key.c_str(),val.c_str()));
            continue;//下发命令与状态相同，不响应
        }

        std::string strcmd = _protocol.get_ctrcmddata(key, val);
        if (strcmd.length() == 0)
            continue;
        //下发命令
        if (send_cmd(strcmd) != (int)strcmd.length())
        {
            LogError(("%s: %s send cmd %s error\r\n", __PRETTY_FUNCTION__, strdevid.c_str(), strcmd.c_str()));
        }
        
        //设置超时时间
        int time  = _protocol.get_retime();
        if (time > 0)
        _stream_ptr->set_timeout(time * 2);//超时时间间隔的两倍
    } 
    return 0;
}

int gateway_client::send_cmd(std::string strcmd)
{
    int ret = _stream_ptr->send((unsigned char*)strcmd.c_str(), strcmd.length());
    if (ret != (int)strcmd.length())
        return ret; //不成功立即返回
    
    //成功立即则更新属性
    std::string strjson = _protocol.get_json_buf();
    if (_strdevid.length() == 0 || strjson.length() == 0)
        return ret;

    event_data *ed = new event_data;
    ed->tasktype = PULISH;
    pulish_data *pdata = new pulish_data;
    pdata->data = new char[strjson.length()];
    strjson.copy(pdata->data, strjson.length(), 0);
    pdata->datalen = strjson.length();
    ed->data = (void*) pdata;
    gateway_taskpool::get_instance()->schedule(boost::static_pointer_cast<gateway_task>(_client_ptr), (void*)ed);
    return ret;
}

int gateway_client::get_current_stamp()
{
    ptime epoch(boost::gregorian::date(1970, boost::gregorian::Jan, 1));
    time_duration time_from_epoch = second_clock::universal_time() - epoch;
    return time_from_epoch.total_seconds();
} 